#include "x_wr_asio.h"
#include "x_errtype.h"
#include "j_module.h"

CWrAsio::CWrAsio()
{
	m_workThread = 0;
	m_bStarted = false;
}

CWrAsio::~CWrAsio()
{

}

int CWrAsio::Init()
{
	if (!m_bStarted)
	{
		for (int i=0; i<IN_FILES; i++)
		{
			m_fds[i].fd = -1;
		}

		m_workThread = 0;
		m_bStarted = true;
		pthread_create(&m_workThread, NULL, WorkThread, this);
		pthread_detach(m_workThread);
	}
	return J_OK;
}

void CWrAsio::Deinit()
{
	if (m_bStarted)
	{
		m_bStarted = false;
		if (m_workThread != 0)
		{
			pthread_cancel(m_workThread);
			m_workThread = 0;
		}
	}
}

int CWrAsio::AddUser(int nSocket, J_AsioUser *pUser)
{
	TLock(m_locker);
	m_asioMap[nSocket] = pUser;

	for (int i=0; i<IN_FILES; i++)
	{
		if (m_fds[i].fd == -1)
		{
			m_fds[i].fd = nSocket;
			m_fds[i].events = POLLOUT;
			break;
		}
	}
	J_OS::LOGINFO("CWrAsio::AddUser poll set insertion sucess fd = %d", nSocket);
	TUnlock(m_locker);

	return J_OK;
}

void CWrAsio::DelUser(int nSocket)
{
	TLock(m_locker);
	AsioMap::iterator it = m_asioMap.find(nSocket);
	if (it != m_asioMap.end())
	{
		m_asioMap.erase(it);
	}

	for (int i=0; i<IN_FILES; i++)
	{
		if (m_fds[i].fd == -1)
		{
			m_fds[i].fd = -1;
			m_fds[i].events = 0;
			break;
		}
	}

	J_OS::LOGINFO("CWrAsio::DelUser epoll set insertion sucess fd = %d", nSocket);
	TUnlock(m_locker);
}

void CWrAsio::OnWork()
{
	int nfds = 0;
	int i = 0;
	int active_fd = 0;
	pthread_setcancelstate(PTHREAD_CANCEL_DEFERRED, NULL);
	while (m_bStarted)
	{
		pthread_testcancel();
		nfds = poll(m_fds, IN_FILES, TIME_DELAY);
		if (nfds <= 0)
		{
			usleep(10);
			continue;
		}

		for (i = 0; i < IN_FILES; i++)
		{
			if (m_fds[i].revents & POLLOUT)
			{
				//read
				TLock(m_locker);
				active_fd = m_fds[i].fd;
				AsioMap::iterator it = m_asioMap.find(active_fd);
				if (it != m_asioMap.end())
				{
					J_AsioUser *pAsioUser = static_cast<J_AsioUser *>(it->second);
					if (pAsioUser)
						pAsioUser->OnWrite(active_fd);
				}
				TUnlock(m_locker);
			}
		}
	}
}
